#pragma once

#include <atomic>
#include <chrono>
#include <condition_variable>
#include <functional>
#include <memory>
#include <mutex>
#include <stdexcept>
#include <string>
#include <thread>
#include <unordered_map>
#include <vector>

#include <boost/date_time/posix_time/posix_time.hpp>

#include "logger.hpp"
#include "mysql_outbox.hpp"

namespace suye
{
class OutboxSyncWorker
{
public:
    using ptr = std::shared_ptr<OutboxSyncWorker>;
    using Handler = std::function<bool(const std::shared_ptr<OutboxEvent>&, std::string&)>;

    OutboxSyncWorker(const OutboxTable::ptr& outbox_table,
                     std::size_t batch_size,
                     std::chrono::milliseconds poll_interval,
                     std::chrono::milliseconds requeue_timeout = std::chrono::milliseconds(30000))
        : _outbox_table(outbox_table)
        , _batch_size(batch_size)
        , _poll_interval(poll_interval)
        , _requeue_timeout(requeue_timeout)
    {
        if (!_outbox_table)
        {
            throw std::invalid_argument("OutboxSyncWorker requires valid OutboxTable instance");
        }
        if (_batch_size == 0)
        {
            _batch_size = 64;
        }
        if (_poll_interval.count() <= 0)
        {
            _poll_interval = std::chrono::milliseconds(500);
        }
    }

    ~OutboxSyncWorker()
    {
        stop();
    }

    void register_handler(const std::string& event_type, const Handler& handler)
    {
        std::lock_guard<std::mutex> lock(_handler_mutex);
        _handlers[event_type] = handler;
    }

    void start()
    {
        bool expected = false;
        if (_running.compare_exchange_strong(expected, true))
        {
            _worker_thread = std::thread([this]() { this->run_loop(); });
        }
    }

    void stop()
    {
        bool expected = true;
        if (_running.compare_exchange_strong(expected, false))
        {
            {
                std::lock_guard<std::mutex> lock(_sleep_mutex);
            }
            _sleep_cv.notify_all();
            if (_worker_thread.joinable())
            {
                _worker_thread.join();
            }
        }
    }

    void run_once()
    {
        process_iteration();
    }

private:
    void run_loop()
    {
        while (_running.load())
        {
            bool has_work = process_iteration();
            if (!has_work)
            {
                std::unique_lock<std::mutex> lock(_sleep_mutex);
                _sleep_cv.wait_for(lock, _poll_interval, [this]() { return !_running.load(); });
            }
        }
    }

    bool process_iteration()
    {
        const auto now = boost::posix_time::second_clock::universal_time();
        try
        {
            const auto timeout = boost::posix_time::milliseconds(static_cast<int>(_requeue_timeout.count()));
            _outbox_table->requeue_stuck_events(timeout, now);
        }
        catch (const std::exception& e)
        {
            LOG_ERROR("OutboxSyncWorker requeue_stuck_events exception: {}", e.what());
        }

        std::vector<std::shared_ptr<OutboxEvent>> events;
        try
        {
            events = _outbox_table->claim_batch(_batch_size, now);
        }
        catch (const std::exception& e)
        {
            LOG_ERROR("OutboxSyncWorker claim_batch exception: {}", e.what());
            return false;
        }

        if (events.empty())
        {
            return false;
        }

        for (const auto& event_ptr : events)
        {
            if (!event_ptr)
            {
                continue;
            }

            std::string error_message;
            bool dispatched = dispatch_event(event_ptr, error_message);
            if (dispatched)
            {
                if (!_outbox_table->mark_processed(event_ptr->event_id()))
                {
                    LOG_ERROR("OutboxSyncWorker mark_processed failed: {}", event_ptr->event_id());
                }
            }
            else
            {
                const auto backoff = boost::posix_time::milliseconds(static_cast<int>(_poll_interval.count()));
                if (!_outbox_table->mark_failed(event_ptr->event_id(), error_message, backoff))
                {
                    LOG_ERROR("OutboxSyncWorker mark_failed failed: {}", event_ptr->event_id());
                }
            }
        }

        return true;
    }

    bool dispatch_event(const std::shared_ptr<OutboxEvent>& event, std::string& error_message)
    {
        Handler handler;
        {
            std::lock_guard<std::mutex> lock(_handler_mutex);
            auto it = _handlers.find(event->event_type());
            if (it != _handlers.end())
            {
                handler = it->second;
            }
        }

        if (!handler)
        {
            LOG_WARN("OutboxSyncWorker 未找到事件处理器: {}", event->event_type());
            return true;
        }

        bool success = false;
        try
        {
            success = handler(event, error_message);
        }
        catch (const std::exception& e)
        {
            error_message = e.what();
            success = false;
        }
        catch (...)
        {
            error_message = "unknown exception";
            success = false;
        }

        if (!success)
        {
            if (error_message.empty())
            {
                error_message = "handler returned false";
            }
            LOG_ERROR("OutboxSyncWorker 处理事件失败: {} - {}", event->event_type(), error_message);
        }
        else
        {
            LOG_DEBUG("OutboxSyncWorker 处理事件成功: {} - {}", event->event_type(), event->event_id());
        }

        return success;
    }

private:
    OutboxTable::ptr _outbox_table;
    std::size_t _batch_size{64};
    std::chrono::milliseconds _poll_interval{500};
    std::chrono::milliseconds _requeue_timeout{30000};

    std::unordered_map<std::string, Handler> _handlers;
    std::mutex _handler_mutex;

    std::atomic<bool> _running{false};
    std::thread _worker_thread;
    std::condition_variable _sleep_cv;
    std::mutex _sleep_mutex;
};
}
